#include <stdio.h>
#include <stdlib.h>
#include <sys/msg.h> 
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <errno.h>
#include <pthread.h>
#include <netdb.h>
#include <list>
#include <set>
#include <string>
#include <unistd.h>
#include	<fcntl.h>

#include "../xml/cxml.h"
#include "../xml/cpuxml.h"
#include "../utilize/timeout.h"
#include "../msgqueue/MsgQue.h"
#include "../sql/sqlite3.h"

extern "C"
{
#include"../noid/lib.h"
}

#include "../kad/node.h"
using DHT::Node;
using DHT::NodeId;
using DHT::Tuple;

#include <iostream>
using namespace std;

/*the server listen on this port   */
#define SERVPORT "2048"
#define DHTPORT 8889
#define dbName "context.db3"

#define dd (65535 - 49152 - 1)
set<int> is; // a set to store used udp socket ports
const static int SQLSIZE = 256;
const char *MSG_FILE = "/";

sqlite3 *pDB = NULL;
char * errMsg = NULL;

struct Arg
{
	int srcType; //最初消息来源，0代表消息队列，1代表socket
	int channel; // 消息队列的队列号或 socket的fd
	struct MsgType *msg; //消息的具体内容指针
};

key_t servKey;
int servMsqId;
int servSfd;
pthread_t msqRecvTid; // thread id for message queue
pthread_t socketRecvTid; // thread id for socket

unsigned char nid[IDLEN];
Node * dhtNode; // for DHT operations

struct addrinfo *host_serv(const char *host, const char *serv, int flags,
		int family, int socktype);
int getTmpPort();
void freePort(int port);
int readReady(int sockFd, int sec, int usec);//实现socket超时功能

void createMsgQue()
{
	servKey = ftok(MSG_FILE, 'a');
	if ((servMsqId = msgget(servKey, S_IRUSR | S_IWUSR | IPC_CREAT | IPC_EXCL))
			== -1)
	{
		servMsqId = msgget(servKey, S_IRUSR | S_IWUSR | IPC_CREAT);
		msgctl(servMsqId, IPC_RMID, 0); // delete it
		servMsqId = msgget(servKey, S_IRUSR | S_IWUSR | IPC_CREAT | IPC_EXCL); //create a new one
	}

	cout << "Server messge queue ID: " << servMsqId << endl;
}

int getASocket(int port, int flags, int family, int socktype)
{
	struct addrinfo *result, *rp;
	int tmpSfd;
	char portTmp[16];

	sprintf(portTmp, "%d", port);
	result = host_serv(NULL, portTmp, flags, family, socktype);
	for (rp = result; rp != NULL; rp = rp->ai_next)
	{
		tmpSfd = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
		if (tmpSfd == -1)
			continue;

		if (bind(tmpSfd, rp->ai_addr, rp->ai_addrlen) == 0)
			break; /* Success */

		close(tmpSfd);
	}

	if (rp == NULL)
	{ /* No address succeeded */
		fprintf(stderr, "Could not bind\n");
		return 0;
	}

	freeaddrinfo(result); /* No longer needed */
	return tmpSfd;
}

void createSocket()
{
	servSfd = getASocket(atoi(SERVPORT), AI_PASSIVE, AF_UNSPEC, SOCK_DGRAM);
	cout << "Server socket fd: " << servSfd << endl;
}

/*
 * 处理从消息队列收到的消息：struct MsgType类型的，分析并调用相关函数，甚至给对方发送回复
 */
void *msgProcess(void *ptr)
{
	struct addrinfo *result;
	vector<Tuple>::iterator ite;
	char reqBuf[MBUFSIZ]; //用来保存发送过来的xml
	char resBuf[MBUFSIZ]; //用来保存准备回复的xml
	bzero(reqBuf, sizeof(reqBuf));
	unsigned char cid[IDLEN];

	int sqlResult = 0;
	int sockResult = 0;

	struct Arg *arg = (struct Arg *) ptr;
	struct MsgType *msg = (struct MsgType *) (arg->msg);
	cout << "Process thread recieve from msgqueue: " << endl << msg->mtext
			<< endl;
	strncpy(reqBuf, msg->mtext, MBUFSIZ);
	delete msg;

	CXML *xml = new CXML();
	xml->readHead(reqBuf);
	//	cout << "Title: " << xml->getTitle() << endl; //ContextRequest
	//	cout << "Type: " << xml->getType() << endl; // ResourceInfo

	//todo 待实现代码							 						//对不同类型的XML生成不同的实例
	if (!strcmp("resourceInfo", xml->getType()))
	{
		free(xml); // 完成使命
		xml = new CPU_XML();

	} else if (!strcmp("linkState", xml->getType()))
	{
		// free(xml);
		//xml = new LINK_XML();
	} else if (!strcmp("interfaceInfo", xml->getType()))
	{
		//free(xml);
		//xml = new IF_XML();
	} else
	{
		free(xml);
		printf("No new xml type match!!!\n");
		return NULL;
	}

	xml->readHead(reqBuf); //再读一次头部
	xml->readDetail(reqBuf); //读出详细内容
	//	cout << "Nid: " << xml->getNID() << endl;
	//	cout << "Oper: " << xml->getOper() << endl; // Store

	xml->formCID(cid);
	//todo ！！！ 待实现函数
	/*
	 *     CID = getCid()           				//得到该上下文信息的CID
	 * 	 	ipList = getRelatedNID()			//为后面的网络存储/查询作准备
	 */


	vector<Tuple> dest = dhtNode->lookUp(NodeId(cid));
	if(dest.empty())
		cout << "shit" << endl;

	cout << "DHT return nodes: " << endl;
	for (ite = dest.begin(); ite != dest.end(); ++ite)
	{
		cout << ite->ip << '\t' << endl;;
	}

	if (!(strcmp("store", xml->getOper()))) // 收到的是存储的请求
	{
		xml->store(pDB); // save();		 	在本地存储一份

		//转发请求，在网络存储n份
		for (ite = dest.begin(); ite != dest.end(); ite++)
		{
			result = host_serv((*ite).ip.c_str(), SERVPORT, AI_CANONNAME, 0, 0);
			if (sendto(servSfd, reqBuf, MBUFSIZ, 0, result->ai_addr,
					result->ai_addrlen) != MBUFSIZ)
				fprintf(stderr, "Error forwarding resquest\n");
			freeaddrinfo(result); /* No longer needed */
		}

	} else if (!(strcmp("query", xml->getOper()))) // 收到的是查询请求query
	{
		sqlResult = xml->query(pDB); //本地查询

		if (sqlResult)
		{
			xml->formXML(resBuf);
		} else //在本地查询不成功
		{
			cout << "No local storage matched... " << endl;
			//转发请求，在网络查询
			int tmpPort = getTmpPort();
			int tmpSfd = getASocket(tmpPort, AI_PASSIVE, AF_UNSPEC, SOCK_DGRAM);
			for (ite = dest.begin(); ite != dest.end(); ite++)
			{
				result = host_serv((*ite).ip.c_str(), SERVPORT, AI_CANONNAME, 0, 0);
				if (sendto(tmpSfd, reqBuf, MBUFSIZ, 0, result->ai_addr,
						result->ai_addrlen) != MBUFSIZ)
					fprintf(stderr, "Error forwarding resquest\n");
				freeaddrinfo(result); /* No longer needed */
			}

			// 从监听的socket里获得回复，得到一个回复即终止(0.5秒之内)
			sockResult = readReady(tmpSfd, 0, 500000);
			if (sockResult)
			{
				recvfrom(tmpSfd, resBuf, MBUFSIZ, 0, NULL, NULL);
				printf("Received response form socket:\n %s\n", resBuf);
			} else
				cout << "*****select timeout: " << endl;

			close(tmpSfd);
			freePort(tmpPort);
		}

		if (sqlResult || sockResult)
		{
			//通过FIFO回复，FIFO的标识在请求xml里有
			char fifoName[32];
			int nwrite;
			int fifoid = xml->getReqID();
			sprintf(fifoName, "/tmp/context%d.fifo", fifoid);
			int fdwrite = open(fifoName, O_WRONLY);
			if (fdwrite)
			{
				cout << "Ready to write: " << fifoName << endl;
				//strcpy(resBuf, "this is a response test");
				nwrite = write(fdwrite, resBuf, MBUFSIZ);
			}
		}

	}

	free(xml);
	return NULL;
}

/*
 * 消息队列监听函数，对于每个接收到的消息，生成一个线程去处理
 */
void *msgQueueRecv(void *ptr)
{

	while (1)
	{
		struct MsgType *msg = new struct MsgType;
		struct Arg *arg = new struct Arg;
		//if servMsgId = 0, receive the oldest msg from the msgQueue, no matter what msgtyp is
		msgrcv(servMsqId, msg, sizeof(struct MsgType) - sizeof(long), 1, 0);

		arg->srcType = 0;
		arg->channel = servMsqId;
		arg->msg = msg;

		pthread_t thread;
		pthread_create(&thread, NULL, &msgProcess, arg);
		pthread_detach(thread);
	}

}

struct ArgSocket
{
	struct sockaddr *srcAddr;
	socklen_t peer_addr_len;
	char *msg; //消息的具体内容指针
};

/*
 * 处理从socket收到的消息，传入参数为string类型，拆包xml，分析并调用相关函数，甚至给对方发送回复
 */
void *sockMsgProcess(void *ptr)
{
	struct ArgSocket *msg = (struct ArgSocket *) ptr;
	char reqBuf[MBUFSIZ]; //用来保存发送过来的xml
	char resBuf[MBUFSIZ]; //用来保存准备回复的xml

	int sqlResult = 0;

	strncpy(reqBuf, (char *) msg->msg, MBUFSIZ);
	cout << "Process thread recieve from socket: " << reqBuf << endl;

	CXML *xml = new CXML();
	xml->readHead(reqBuf);
	// 待实现代码							 						//对不同类型的XML生成不同的实例
	if (!strcmp("resourceInfo", xml->getType()))
	{
		free(xml); // 完成使命
		xml = new CPU_XML();

	} else if (!strcmp("linkState", xml->getType()))
	{
		// free(xml);
		//xml = new LINK_XML();
	} else if (!strcmp("interfaceInfo", xml->getType()))
	{
		//free(xml);
		//xml = new IF_XML();
	} else
	{
		free(xml);
		printf("No new xml type match!!!\n");
		return NULL;
	}
	xml->readHead(reqBuf); //再读一次头部
	xml->readDetail(reqBuf); //读出详细内容


	if (!(strcmp("store", xml->getOper()))) // 收到的是存储的请求
	{
		xml->store(pDB); // save();		 	在本地存储一份
	} else if (!(strcmp("query", xml->getOper()))) // 收到的是查询请求query
	{
		sqlResult = xml->query(pDB); //本地查询

		if (sqlResult)
		{
			xml->formXML(resBuf);
			int nResonse = strlen(resBuf) + 1;
			if (sendto(servSfd, resBuf, nResonse, 0, msg->srcAddr, //利用daemon的servSfd把回复发送出去
					msg->peer_addr_len) != nResonse)
				fprintf(stderr, "Error sending response\n");
		} else
			cout << "No local storage matched... " << endl;

	}

	free(xml);
	return NULL;
}

/*
 * socket监听函数，对于每个接收到的消息，生成一个线程去处理
 * 传出参数为string类型，要用new来生成
 */
void *socketRecv(void *ptr)
{
	char buf[MBUFSIZ];
	socklen_t peer_addr_len;
	ssize_t nread;
	int s;
	while (1)
	{
		struct sockaddr_storage peer_addr;
		peer_addr_len = sizeof(struct sockaddr_storage);
		bzero(buf, sizeof(buf));
		nread = recvfrom(servSfd, buf, MBUFSIZ, 0,
				(struct sockaddr *) &peer_addr, &peer_addr_len);

		char host[NI_MAXHOST], service[NI_MAXSERV];
		s = getnameinfo((struct sockaddr *) &peer_addr, peer_addr_len, host,
				NI_MAXHOST, service, NI_MAXSERV, NI_NUMERICSERV);
		if (s == 0)
			printf("Received %ld bytes from %s:%s\n", (long) nread, host,
					service);
		else
			fprintf(stderr, "getnameinfo: %s\n", gai_strerror(s));
		if (nread == -1)
			continue; /* Ignore failed request */

		struct ArgSocket *arg = new struct ArgSocket;
		arg->srcAddr = (struct sockaddr *) &peer_addr;
		arg->peer_addr_len = peer_addr_len;
		arg->msg = buf;

		pthread_t thread;
		pthread_create(&thread, NULL, &sockMsgProcess, arg);
		pthread_detach(thread);
	}

}

int main(void)
{
	//DHT
	//	ID id;
	//	int result;

	//get node id


	if (GetConfigValue("./config.ini", "HOST", "NODEID", nid, STRSTR) != 0)
	{
		perror("Failed to read NODEID from configuration file.\n");
		return -1;
	}
	//get local address and dht "well known node" address
	char localAddr[64];
	char wknAddr[64];
	if (GetConfigValue("./config.ini", "DHT", "LOCALADDR", localAddr, STRSTR)
			!= 0)
	{
		perror("Failed to read LOCALADDR from configuration file.\n");
		return -1;
	}
	if (GetConfigValue("./config.ini", "DHT", "WKNADDR", wknAddr, STRSTR) != 0)
	{
		perror("Failed to read WKNADDR from configuration file.\n");
		return -1;
	}
	cout << "Init ... " << endl << "NodeId: " << nid << endl << "LocalAddr: "
			<< localAddr << endl << "WknAddr: " << wknAddr << endl;

	dhtNode = new Node(NodeId((unsigned char *) nid), localAddr, DHTPORT);
	dhtNode->bootsTrap(wknAddr, 8889);

	//打开一个数据库，如果改数据库不存在，则创建一个名字为databaseName的数据库文件
	int rc = sqlite3_open(dbName, &pDB);
	if (rc)
	{
		cout << " Open the database " << dbName << " failed" << endl;
		return (-1);
	}

	createMsgQue(); //~~~~~~~~~~~~~~~~
	createSocket(); // ~~~~~~~~~~~~~~~

	pthread_create(&msqRecvTid, NULL, &msgQueueRecv, NULL); //创建消息队列接收线程
	pthread_create(&socketRecvTid, NULL, &socketRecv, NULL); //创建socket线程，用以接收别的结点传递过来的请求

	pthread_join(msqRecvTid, NULL); // pthread_join应该放在全部线程创建以后
	pthread_join(socketRecvTid, NULL);

	sqlite3_close(pDB);
	return 0;
}

/* getaddrinfo() returns a list of address structures.
 Try each address until we successfully connect(2).
 If socket(2) (or connect(2)) fails, we (close the socket
 and) try the next address. */
struct addrinfo *host_serv(const char *host, const char *serv, int flags,
		int family, int socktype)
{
	int n;
	struct addrinfo hints, *res;

	bzero(&hints, sizeof(struct addrinfo));
	//hints.ai_flags = AI_CANONNAME; //always return canonical name
	hints.ai_flags = flags;
	hints.ai_family = family; //AF_UNSPEC, AF_INET, AF_INET6, etc
	hints.ai_socktype = socktype; //0, SOCK_STREAM, SOCK_DGRAM, etc

	if ((n = getaddrinfo(host, serv, &hints, &res)) != 0)
		return NULL;

	return (res);
}

/*
 * 随机生成一个可用的临时端口值，范围是49152至65535
 * 放入set<int> is;
 */
int getTmpPort()
{
	pair<set<int>::iterator, bool> Ret;
	srand(time(0));
	int tmp;
	while (1)
	{
		tmp = rand() % dd;
		Ret = is.insert(tmp);
		if (!(Ret.second))
			continue;
		else
			break;
	}
	tmp += 49152;

	return tmp;
}

/*
 * 从set<int> is里释放指定的临时端口值
 */
void freePort(int port)
{
	is.erase(port);
}

